package ooo.reindeer.concurrent.netqueue.handler;

import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import ooo.reindeer.commons.Property;
import ooo.reindeer.concurrent.BlockingFileQueue;
import ooo.reindeer.concurrent.coder.BytesCoder;
import ooo.reindeer.concurrent.netqueue.message.Command;
import ooo.reindeer.concurrent.netqueue.message.Message;
import ooo.reindeer.logging.ILogger;
import ooo.reindeer.logging.LogUtil;

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executor;
import java.util.concurrent.Executors;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

public class QueueClientMessageHandler extends SimpleChannelInboundHandler<Message> {

    private static final ILogger logger = LogUtil.getLogger(QueueClientMessageHandler.class);
    private static final ConcurrentHashMap<String, BlockingQueue<byte[]>> queueMap = new ConcurrentHashMap<String, BlockingQueue<byte[]>>();
    private static final ConcurrentHashMap<ChannelHandlerContext, Executor> executorMap = new ConcurrentHashMap<ChannelHandlerContext, Executor>();
    private static final ConcurrentHashMap<String, MessagePushHandler> messagePushHandlerMap = new ConcurrentHashMap<String, MessagePushHandler>();
    private static Lock topicLock = new ReentrantLock();
    private static Lock executorLock = new ReentrantLock();

    @Override
    public void channelActive(ChannelHandlerContext ctx) {
        logger.info("queue client active [{}]", ctx);
        ctx.fireChannelActive();
    }

    @Override
    public void channelInactive(ChannelHandlerContext ctx) {
        logger.info("queue client Inactive [{}]", ctx);
        ctx.fireChannelInactive();
    }

    @Override
    protected void channelRead0(ChannelHandlerContext channelHandlerContext, Message message) throws Exception {

        if (message.getCommand() == Command.PUT) {
            logger.debug("PUT {}", message);
            BlockingQueue<byte[]> queue = getQueue(message);
            queue.put(message.getData());
            Message answer = new Message();
            answer.setCommand(Command.PUT_ANSWER);
            answer.setTopic(message.getTopic());
            answer.setStatus("SUCCE");
            channelHandlerContext.writeAndFlush(answer);
        } else if (message.getCommand() == Command.REGISTE) {
            logger.debug("REGISTE {}", message);
            Executor executor = getExecutor(channelHandlerContext);
            MessagePushHandler messagePushHandler = new MessagePushHandler(message, channelHandlerContext, getQueue(message), messagePushHandlerMap);
            executor.execute(messagePushHandler);

            Message registeAnswer = new Message();
            registeAnswer.setCommand(Command.REGISTE_ANSWER);
            registeAnswer.setTopic(message.getTopic());
            registeAnswer.setStatus("SUCCE");
            channelHandlerContext.writeAndFlush(registeAnswer);
        } else if (message.getCommand() == Command.CANCEL_REGISTE) {
            logger.debug("REGISTE {}", message);

            MessagePushHandler messagePushHandler = messagePushHandlerMap.remove(String.format("%s-%s", channelHandlerContext, message.getTopic()));
            if (messagePushHandler != null) {
                messagePushHandler.exit();
            }
            Message registeAnswer = new Message();
            registeAnswer.setCommand(Command.CANCEL_REGISTE_ANSWER);
            registeAnswer.setTopic(message.getTopic());
            registeAnswer.setStatus("SUCCE");
            channelHandlerContext.writeAndFlush(registeAnswer);
        } else if (message.getCommand() == Command.PUSH_ANSWER) {
            logger.debug("PUSH_ANSWER {}", message);
        }

    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        logger.error("exception [{}] :{}", ctx, cause);
        ctx.close();
    }

    private Executor getExecutor(ChannelHandlerContext key) throws Exception {
        Executor executor = executorMap.get(key);

        if (executor == null) {
            executorLock.lock();
            try {
                executor = executorMap.get(key);
                if (executor == null) {
                    executor = Executors.newCachedThreadPool();
                    executorMap.put(key, executor);
                }
            } finally {
                executorLock.unlock();
            }
        }
        return executor;
    }

    private BlockingQueue<byte[]> getQueue(Message message) throws Exception {
        BlockingQueue<byte[]> queue = queueMap.get(message.getTopic());

        if (queue == null) {
            topicLock.lock();
            try {
                queue = queueMap.get(message.getTopic());
                if (queue == null) {
                    queue = new BlockingFileQueue<byte[]>(Property.get("rootPath", "./") + "/" + message.getTopic(),
                            Property.get("blockSize", 4 * 1024 * 1024), new BytesCoder());
                    queueMap.put(message.getTopic(), queue);
                }
            } finally {
                topicLock.unlock();
            }
        }
        return queue;
    }

}
